package mapreduce;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;

public class MapDiskManager<KEYOUT, VALUEOUT> extends DiskManager<KEYOUT, VALUEOUT> {

	private int numofreduce;
	private int Taskid;
	BufferedWriter[] bws;

	public MapDiskManager(String jobdir, int taskid, int numofreduce)
			throws IOException {
		super(jobdir);
		this.numofreduce = numofreduce;
		this.Taskid = taskid;
		bws = new BufferedWriter[this.numofreduce];
		for (int i = 0; i < numofreduce; i++) {
			bws[i] = new BufferedWriter(new FileWriter(this.Jobdir
					+ "reduce/input/" + i + "/" + Taskid + ".data", true));
		}
	}

	private int hash_key(String str) {

		return Math.abs(str.hashCode()) % numofreduce;
	}
    @Override
	public void write(KEYOUT key, VALUEOUT value)  {
		int hashkey;
		hashkey = hash_key(key.toString());
		try {
			bws[hashkey].write(key + "\t" + value);
			bws[hashkey].newLine();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

	@Override
	public void close()  {
		try {
			for (int i = 0; i < numofreduce; i++) 
				bws[i].close();
			
		} catch (Exception e) {
			// TODO: handle exception
		}
		
		}
	

}
